package com.kitten.search.biz.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.common.collect.Maps;
import com.kitten.search.biz.config.CanalProperties;
import com.kitten.search.biz.domain.mapper.SelectMapper;
import com.kitten.search.biz.enums.NoteStatusEnum;
import com.kitten.search.biz.enums.NoteVisibleEnum;
import com.kitten.search.biz.enums.UserStatusEnum;
import com.kitten.search.biz.index.NoteIndex;
import com.kitten.search.biz.index.UserIndex;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;

import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;


import java.io.IOException;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.List;

/**
 * @author kitten
 */
@Slf4j
@Component
public class CanalSchedule implements Runnable {
    @Resource
    private CanalProperties canalProperties;
    @Resource
    private CanalConnector canalConnector;
    @Resource
    private RestHighLevelClient restHighLevelClient;
    @Resource
    private SelectMapper selectMapper;

    @Override
    @Scheduled(fixedDelay = 100)
    public void run() {
        long batchId = -1; // 初始化批次id

        try {
            Message message = canalConnector.getWithoutAck(canalProperties.getBatchSize());
            batchId = message.getId(); // 批次id
            long batchSize = message.getEntries().size(); // 当前批次的数据条目
            if (batchId == -1 && batchSize == 0) {
                try {   // 拉取数据为空，休眠1秒防止频繁拉取
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            } else {
                processEntry(message.getEntries());
            }
            // Ack确认
            canalConnector.ack(batchId);
        } catch (Exception e) {
            log.error("消费批次异常: {}", e);
            canalConnector.rollback(batchId);
        }
    }


    /**
     * 处理批次的数据条目
     *
     * @param entries
     */
    private void processEntry(List<CanalEntry.Entry> entries) throws Exception {
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) { // 处理行数据
                CanalEntry.EventType eventType = entry.getHeader().getEventType();
                String schemaName = entry.getHeader().getSchemaName();
                String tableName = entry.getHeader().getTableName();
                // RowChange对象, 包含 rowData 和 事件 相关信息
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());

                // 处理每行数据
                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    List<CanalEntry.Column> afterColumns = rowData.getAfterColumnsList();
                    if (afterColumns.size() == 0) { // 手动删除数据库, afterColumns 为空, 做记录, 这里不处理
                        continue;
                    }
                    Map<String, Object> columnMap = parseColumns2Map(afterColumns);

                    log.info("自定义处理 ======= ");
                    processEvent(columnMap, tableName, eventType);

                }
            }
        }
    }

    private Map<String, Object> parseColumns2Map(List<CanalEntry.Column> columns) {
        Map<String, Object> map = Maps.newHashMap();
        columns.forEach(column -> {
            if (Objects.isNull(column)) {
                return;
            }
            map.put(column.getName(), column.getValue());
        });
        return map;
    }

    private void processEvent(Map<String, Object> columnMap, String tableName, CanalEntry.EventType eventType) throws Exception {
        switch (tableName) {
            case "t_note" -> {
                handleNoteEvent(columnMap, eventType);
            }
            case "t_user" -> {
                handleUserEvent(columnMap, eventType);
            }
            default -> {
                log.warn("不支持处理的表: {}", tableName);
            }
        }
    }

    /**
     * 处理笔记事件
     *
     * @param columnMap
     * @param eventType
     */
    private void handleNoteEvent(Map<String, Object> columnMap, CanalEntry.EventType eventType) throws Exception {
        long noteId = Long.parseLong(columnMap.get("id").toString());
        switch (eventType) {    // 处理事件类型
            case INSERT -> {
                syncNoteIndex(noteId);
            }
            case UPDATE -> {
                Integer status = Integer.parseInt(columnMap.get("status").toString());  // 笔记状态
                Integer visible = Integer.parseInt(columnMap.get("visible").toString());    // 笔记可见性

                if (Objects.equals(status, NoteStatusEnum.NORMAL.getCode()) && Objects.equals(visible, NoteVisibleEnum.PUBLIC.getCode())) {
                    syncNoteIndex(noteId);  // 同步笔记到es文档
                } else if (Objects.equals(visible, NoteVisibleEnum.PRIVATE.getCode())
                        || Objects.equals(status, NoteStatusEnum.DOWNED.getCode())
                        || Objects.equals(status, NoteStatusEnum.DELETE.getCode())) {
                    // TODO: 笔记不可见或状态异常，删除对应es文档
                    deleteNoteDocument(String.valueOf(noteId));
                }

            }
            default -> {
                log.warn("不支持处理的事件类型: {}", eventType);
            }
        }
    }

    // 同步笔记到es文档
    private void syncNoteIndex(Long noteId) throws Exception {
        List<Map<String, Object>> result = selectMapper.selectEsNoteIndexData(noteId, null);
        for (Map<String, Object> map : result) {
            IndexRequest indexRequest = new IndexRequest(NoteIndex.NAME);
            indexRequest.id(String.valueOf(map.get(NoteIndex.FIELD_NOTE_ID)));
            indexRequest.source(map);   // 设置文档内容
            restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);   // 同步到es
        }
    }

    // 删除指定es文档
    private void deleteNoteDocument(String noteId) throws IOException {
        DeleteRequest deleteRequest = new DeleteRequest(NoteIndex.NAME, noteId);
        restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
    }

    /**
     * 处理用户事件
     *
     * @param columnMap
     * @param eventType
     */
    private void handleUserEvent(Map<String, Object> columnMap, CanalEntry.EventType eventType) throws Exception {
        log.info("处理用户事件: {}, {}", columnMap, eventType);
        Long userId = Long.parseLong(columnMap.get("id").toString());
        switch (eventType) {
            case INSERT -> {
                syncUserIndex(userId);
            }
            case UPDATE -> {
                Integer status = Integer.parseInt(columnMap.get("status").toString());
                Integer isDeleted = Integer.parseInt(columnMap.get("is_deleted").toString());
                if (Objects.equals(status, UserStatusEnum.ENABLED.getCode()) && Objects.equals(isDeleted, 0)) {  // 用户状态为 启用 且 未删除
                    syncUserIndexAndNoteIndex(userId);  // 同步用户索引 以及对应 笔记索引
                } else if (Objects.equals(status, UserStatusEnum.DISABLED.getCode()) || Objects.equals(isDeleted, 1)) {   // 用户状态为 禁用 或 已删除
                    // 删除用户文档
                    deleteUserDocument(String.valueOf(userId));
                }
            }
            default -> {
                log.info("不支持处理的事件类型 for user: {}", eventType);
            }
        }
    }

    // 同步用户到es文档
    private void syncUserIndex(Long userId) throws Exception {
        List<Map<String, Object>> result = selectMapper.selectEsUserIndexData(userId);
        for (Map<String, Object> map : result) {
            IndexRequest indexRequest = new IndexRequest(UserIndex.NAME);
            indexRequest.id(String.valueOf(map.get(UserIndex.FIELD_USER_ID)));
            indexRequest.source(map);   // 设置文档内容
            restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);   // 同步到es
        }
    }
    // 同步用户文档 以及对应 笔记文档
    private void syncUserIndexAndNoteIndex(Long userId) throws Exception {
        BulkRequest bulkRequest = new BulkRequest();
        // 用户索引
        List<Map<String, Object>> userResult = selectMapper.selectEsUserIndexData(userId);
        for (Map<String, Object> userMap : userResult) {
            IndexRequest userIndexRequest = new IndexRequest(UserIndex.NAME);
            userIndexRequest.id(String.valueOf(userMap.get(UserIndex.FIELD_USER_ID)));
            userIndexRequest.source(userMap);   // 设置用户文档内容
            bulkRequest.add(userIndexRequest);
        }
        // 笔记索引
        List<Map<String, Object>> noteResult = selectMapper.selectEsNoteIndexData(null, userId);
        for (Map<String, Object> noteMap : noteResult) {
            IndexRequest noteIndexRequest = new IndexRequest(NoteIndex.NAME);
            noteIndexRequest.id(String.valueOf(noteMap.get(NoteIndex.FIELD_NOTE_ID)));
            noteIndexRequest.source(noteMap);   // 设置笔记文档内容
            bulkRequest.add(noteIndexRequest);
        }
        restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);   // 批量同步到es
    }

    // 删除指定用户文档
    private void deleteUserDocument(String userId) throws IOException {
        DeleteRequest deleteRequest = new DeleteRequest(UserIndex.NAME, userId);
        restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
    }
}